package com.cenntro.common.mq.consumer.processor;

import java.util.Arrays;
import java.util.List;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import com.cenntro.common.redis.IRedisService;
import com.cenntro.common.util.KeyUtil;
/**
 * 所有消息处理继承该类
 */
public abstract class AbstractMQMsgProcessor implements MQMsgProcessor{
	
	protected static final Logger logger = LoggerFactory.getLogger(AbstractMQMsgProcessor.class);
    @Autowired
    private IRedisService redisService;
    
	@Override
	public MQConsumeResult handle(String topic, String tag, List<MessageExt> msgs) {
		MQConsumeResult mqConsumeResult = new MQConsumeResult();
		/**可以增加一些其他逻辑*/
		for (MessageExt messageExt : msgs) {
			//消费具体的消息，抛出钩子供真正消费该消息的服务调用
			String key = KeyUtil.keyOf("REPEAT", topic, messageExt.getMsgId());
			if(!this.repeatable()){
				String msgId = redisService.get(key, String.class);
				if (StringUtils.isNotBlank(msgId)){
					logger.info("忽略重复mq消息msgId={}, topic ={}, tag={}", msgId, messageExt.getTopic(), messageExt.getTags() );
					mqConsumeResult.setSuccess(true);
					continue;
				}
			}
			try{
				mqConsumeResult = this.consumeMessage(tag,messageExt.getKeys()==null?null:Arrays.asList(messageExt.getKeys().split(MessageConst.KEY_SEPARATOR)),messageExt);
			}finally{
				redisService.set(key, messageExt.getMsgId(), 24*60*60*5);
			}
		}
		/**可以增加一些其他逻辑*/
		return mqConsumeResult;
	}
	/**
	 * 消息某条消息
	 */
	protected abstract MQConsumeResult consumeMessage(String tag,List<String> keys, MessageExt messageExt);
	/**
	 * 不可重复
	 */
	public  Boolean repeatable(){
		return false;
	}
}
